package com.abc.base.spark.api;

import com.abc.base.common.model.HouseInfo;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.*;
import com.abc.base.spark.config.SparkConfig;

import java.util.List;

public class ProcessingData {


    public List<HouseInfo> processingData() {
        SparkSession spark = SparkConfig.getSparkSession();
        String mysqlUrl = "jdbc:mysql://node1:3306/test";
        String query = "SELECT * FROM house_info where id > 3000";


        Dataset<Row> df = spark.read()
                .format("jdbc")
                .option("url", mysqlUrl)
                .option("dbtable", "house_info")  // 只读取表名
                .option("user", "root")
                .option("password", "123456")
                .option("driver", "com.mysql.cj.jdbc.Driver")
                .load();

        Dataset<Row> filteredDf = df.filter(df.col("id").gt(3000));

        Dataset<HouseInfo> houseInfoDataset = filteredDf.map((MapFunction<Row, HouseInfo>) row -> {
            HouseInfo houseInfo = new HouseInfo();
            houseInfo.setId(row.getInt(0));
            houseInfo.setTitle(row.getString(1));
            houseInfo.setAddress(row.getString(2));
            houseInfo.setHouseInfo(row.getString(3));
            houseInfo.setFollowInfo(row.getString(4));
            houseInfo.setTotalPrice(row.getString(5));
            houseInfo.setItemPrice(row.getString(6));
            return houseInfo;
        }, Encoders.bean(HouseInfo.class));

        Dataset<Row> processedDF = df
                .withColumn("floor", functions.expr("regexp_extract(house_info, '([^\\\\(]+)\\\\(共\\\\d+层', 1)"))
                .withColumn("total_floors", functions.expr("regexp_extract(house_info, '共(\\\\d+)层', 1)"))
                .withColumn("layout", functions.expr("regexp_extract(house_info, '(\\\\d+室\\\\d+厅)', 1)"))
                .withColumn("area", functions.expr("cast(regexp_extract(house_info, '(\\\\d+)平米', 1) as float)"))
                .withColumn("direction", functions.expr("regexp_extract(house_info, '(朝\\\\w+)', 1)"))
                .withColumn("followers", functions.expr("cast(regexp_extract(follow_info, '(\\\\d+)人关注', 1) as int)"))
                .withColumn("days_ago", functions.expr("cast(regexp_extract(follow_info, '(\\\\d+)天前发布', 1) as int)"))
                .withColumn("total_price", functions.expr("cast(regexp_replace(total_price, '万', '') as float)"))
                .withColumn("item_price", functions.expr("cast(regexp_replace(item_price, '元/平', '') as float)"));

        processedDF = processedDF.withColumn("id", functions.monotonically_increasing_id());

        processedDF.write()
                .format("jdbc")
                .option("url", mysqlUrl)
                .option("dbtable", "house_info_new")
                .option("user", "root")
                .option("password", "123456")
                .option("driver", "com.mysql.cj.jdbc.Driver")
                .mode("overwrite") // 如果存在则覆盖
                .save();

        return houseInfoDataset.collectAsList();
    }
}
